package com.n9.buymsg.service.qqonline;

import com.n9.buymsg.AppContext;
import com.n9.buymsg.event.LogMsgViewEvent;
import com.n9.buymsg.event.QQMsgCountEvent;
import com.n9.sdk.NineIP;
import com.n9.sdk.model.PurchaseMessage;
import com.n9.sdk.response.DefaultResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 远程存储service
 * @Author: ice
 * @Date: 2020-07-06 11:39
 */
public class RemoteStorePurchaseMessageService implements PurchaseMessageService {
    private static final Logger LOGGER = LoggerFactory.getLogger(RemoteStorePurchaseMessageService.class);

    private static final LinkedBlockingQueue<PurchaseMessage> MESSAGES = new LinkedBlockingQueue<>();
    private static final Executor EXECUTOR = new ScheduledThreadPoolExecutor(5);

    private static volatile boolean startFlag = true;
    private static final int STORE_FAILED_RETRY_MAX_TIMES = 3;//保存失败重试最大次数

    private static final Map<Long, AtomicLong> BUY_MSG_COUNTER = new ConcurrentHashMap<>();//求购消息计数器

    static {
        start();
    }

    private static void start() {
        LOGGER.info("========RemoteStorePurchaseMessageService starting========");
        new Thread(()->{
            LOGGER.info("========RemoteStorePurchaseMessageService running========");
            while (startFlag) {
                try {
                    PurchaseMessage msg = MESSAGES.take();
                    EXECUTOR.execute( ()-> storeMsgToRemote(msg));
                } catch (InterruptedException e) {
                    LOGGER.error("求购消息保存失败:{}", e.getMessage());
                }
            }
        }).start();
    }

    private static void storeMsgToRemote(PurchaseMessage msg) {
        boolean storeResult = false;
        for (int i = 1; i <= STORE_FAILED_RETRY_MAX_TIMES; i++) {
            try {
                NineIP nineIP = NineIPFactory.create();
                DefaultResponse defaultResponse = nineIP.submitPurchaseMsg(msg);
                if (defaultResponse.isOk()) {
                    storeResult = true;
                    break;
                } else {
                    LOGGER.error("消息{}第{}次存储失败，远程服务器返回错误：{}", msg.getMsgId(), i, defaultResponse.getMsg());
                }
            } catch (Exception e) {
                LOGGER.error("消息{}第{}次存储异常,异常消息：{}", msg.getMsgId(), i, e.getMessage());
            }
        }
        if (!storeResult) {
            String logMsg = String.format("QQ:%s的消息:%d,存储失败,转入处理失败流程", msg.getReceiver(), msg.getMsgId());
            LOGGER.error(logMsg);
            AppContext.getInstance().publishEvent(new LogMsgViewEvent(logMsg));
            MsgStoreFailedHandler.handleFailed(msg);
        } else {
            countPurchaseMessage(msg);
        }
    }

    private static void countPurchaseMessage(PurchaseMessage msg) {
        long qqUin = Long.parseLong(msg.getReceiver());

        AtomicLong atomicLong = BUY_MSG_COUNTER.get(qqUin);
        if (Objects.isNull(atomicLong)) {
            atomicLong = new AtomicLong(0);
            BUY_MSG_COUNTER.put(qqUin, atomicLong);
        }
        long curCount = atomicLong.incrementAndGet();
        AppContext.getInstance().publishEvent(QQMsgCountEvent.createBuyMsgEvent(qqUin, curCount));
    }

    @Override
    public void save(PurchaseMessage purchaseMessage) {
        try{
            int cacheMsgSize = MESSAGES.size();
            if (cacheMsgSize > 500) {
                AppContext.getInstance().publishEvent(new LogMsgViewEvent(String.format("求购消息缓存即将溢出,请联系技术人员，当前缓存数量:%d", cacheMsgSize)));
            }
            MESSAGES.put(purchaseMessage);
        } catch (InterruptedException e) {
            MsgStoreFailedHandler.handleFailed(purchaseMessage);
            LOGGER.error("求购消息保存失败:{}", e.getMessage());
        }
    }
}
